﻿
using System;
using System.Collections.Generic;
using System.IO;
using MX.Core.Caching.Redis.Configuration;


namespace MX.Core.Caching.Redis.Sider
{
	// A reader that's a bit more aware of the redis protocol than RedisReader
	public class ProtocolReader
	{
		 
		readonly static RedisReader _reader = new RedisReader ();
		readonly static ProtocolEncoder _encoder = new ProtocolEncoder ();

		public ProtocolReader (
			 )
		{
			 
		}


		public int ReadInt (Stream _stream)
		{
			readType (_stream, ResponseType.Integer);
			return _reader.ReadNumberLine (_stream);
		}

		public long ReadInt64 (Stream _stream)
		{
			readType (_stream, ResponseType.Integer);
			return _reader.ReadNumberLine64 (_stream);
		}

		public double ReadDouble (Stream _stream)
		{
			return _encoder.DecodeDouble (readBulk (_stream));
		}

		public double? ReadDoubleOrNull (Stream _stream)
		{
			var bulk = readBulkOrNull (_stream);
			return bulk.HasValue ? (double?)_encoder.DecodeDouble (bulk.Value) : null;
		}

		public bool ReadBool (Stream _stream)
		{
			readType (_stream, ResponseType.Integer);
			return _reader.ReadNumberLine (_stream) == 1;
		}

		public DateTime ReadDateTime (Stream _stream)
		{
			readType (_stream, ResponseType.Integer);
			return _encoder.DecodeDateTime (_reader.ReadNumberLine64 (_stream));
		}

		public TimeSpan ReadTimeSpan (Stream _stream)
		{
			readType (_stream, ResponseType.Integer);
			return _encoder.DecodeTimeSpan (_reader.ReadNumberLine64 (_stream));
		}

		public string ReadStatus (Stream _stream)
		{
			readType (_stream, ResponseType.SingleLine);
			return _reader.ReadStatusLine (_stream);
		}

		public bool ReadStatus (Stream _stream, string expectedMessage)
		{
			return ReadStatus (_stream) == expectedMessage;
		}

		public bool ReadOk (Stream _stream)
		{
			return ReadStatus (_stream, "OK");
		}

		public bool ReadQueued (Stream _stream)
		{
			return ReadStatus (_stream, "QUEUED");
		}


		public string ReadStrBulk (Stream _stream)
		{
			var bulk = readBulkOrNull (_stream);
			return bulk.HasValue ? _encoder.DecodeStr (bulk.Value) : null;
		}

		public byte[] ReadRawBulk (Stream _stream)
		{
			readType (_stream, ResponseType.Bulk);

			var length = _reader.ReadNumberLine (_stream);
			return length < 0 ? null : _reader.ReadBulk (_stream, length);
		}

		public int ReadStreamedBulk (Stream _stream, Stream stream)
		{
			readType (_stream, ResponseType.Bulk);

			var length = _reader.ReadNumberLine (_stream);
			if (length > -1)
				_reader.ReadBulkTo (_stream, stream, length);

			return length;
		}

		public byte[] ReadSerializedBulk (Stream _stream)
		{
			readType (_stream, ResponseType.Bulk);
			var length = _reader.ReadNumberLine (_stream);
			return length < 0 ?
				null : _reader.ReadSerializedBulk (_stream, length);
		}

		public string[] ReadStrMultiBulk (Stream _stream)
		{
			readType (_stream, ResponseType.MultiBulk);

			var count = _reader.ReadNumberLine (_stream);
			if (count == -1)
				return null;

			var result = new string[count];
			for (var i = 0; i < result.Length; i++)
				result [i] = ReadStrBulk (_stream);

			return result;
		}

		public byte[][] ReadSerializedMultiBulk (Stream _stream)
		{
			readType (_stream, ResponseType.MultiBulk);

			var count = _reader.ReadNumberLine (_stream);
			if (count == -1)
				return null;

			var result = new byte[count][];
			for (var i = 0; i < result.Length; i++)
				result [i] = ReadSerializedBulk (_stream);

			return result;
		}

		public int ReadMultiBulkHeader (Stream _stream)
		{
			readType (_stream, ResponseType.MultiBulk);
			return _reader.ReadNumberLine (_stream);
		}


		public KeyValuePair<string, string>? ReadStrKeyValue (Stream _stream)
		{
			var arr = ReadStrMultiBulk (_stream);
			return arr == null ?
        (KeyValuePair<string, string>?)null :
        new KeyValuePair<string, string> (arr [0], arr [1]);
		}

		public KeyValuePair<string, byte[]>? ReadSerializedKeyValue (Stream _stream)
		{
			readType (_stream, ResponseType.MultiBulk);

			var count = _reader.ReadNumberLine (_stream);
			if (count == -1)
				return null;

			return new KeyValuePair<string, byte[]> (
				ReadStrBulk (_stream),
				ReadSerializedBulk (_stream));
		}

		public KeyValuePair<string, string>[] ReadStrKeyValues (Stream _stream)
		{
			readType (_stream, ResponseType.MultiBulk);

			var count = _reader.ReadNumberLine (_stream);
			if (count == -1)
				return null;

			var result = new KeyValuePair<string, string>[count / 2];
			for (var i = 0; i < result.Length; i++) {
				result [i] = new KeyValuePair<string, string> (
					ReadStrBulk (_stream),
					ReadStrBulk (_stream));
			}

			return result;
		}

		public KeyValuePair<string, byte[]>[] ReadSerializedKeyValues (Stream _stream)
		{
			readType (_stream, ResponseType.MultiBulk);

			var count = _reader.ReadNumberLine (_stream);
			if (count == -1)
				return null;

			var result = new KeyValuePair<string,byte[]>[count / 2];
			for (var i = 0; i < result.Length; i++) {
				result [i] = new KeyValuePair<string, byte[]> (
					ReadStrBulk (_stream),
					ReadSerializedBulk (_stream));
			}

			return result;
		}

		public KeyValuePair<byte[], double>[] ReadSerializedWithScores (Stream _stream)
		{
			readType (_stream, ResponseType.MultiBulk);

			var count = _reader.ReadNumberLine (_stream);
			if (count == -1)
				return null;

			var result = new KeyValuePair<byte[], double>[count / 2];
			for (var i = 0; i < result.Length; i++) {
				result [i] = new KeyValuePair<byte[], double> (
					ReadSerializedBulk (_stream),
					ReadDouble (_stream));
			}

			return result;
		}


		public Message  ReadMessage (Stream _stream)
		{
			readType (_stream, ResponseType.MultiBulk);
			var count = _reader.ReadNumberLine (_stream);

			var type = MessageTypes.Parse (ReadStrBulk (_stream));

			switch (type) {
				case MessageType.Message:
					{
						return Message.Create (type, null, ReadStrBulk (_stream), ReadSerializedBulk (_stream));
					}
				case MessageType.PMessage:
					{
						return Message.Create (type, ReadStrBulk (_stream), ReadStrBulk (_stream),
							ReadSerializedBulk (_stream));
					}

				case MessageType.Unsubscribe:
				case MessageType.Subscribe:
					{
						return Message.Create (type, null, ReadStrBulk (_stream), null, ReadInt (_stream));
					}

				case MessageType.PUnsubscribe:
				case MessageType.PSubscribe:
					{
						var pattern = ReadStrBulk (_stream);
						return Message.Create (type, pattern, null, null, ReadInt (_stream));
					}

				case MessageType.Unknown:
				default:
					{

						// maintain protocol even when message type is unknown
						// by reading out all the bulks redis told us is available
						count--; // type field already read
						for (var i = 0; i < count; i++)
							ReadRawBulk (_stream);

						return Message.Create (type, null, null, null);
					}
			}
		}


		private void readType (Stream _stream, ResponseType expectedType)
		{
			var type = _reader.ReadTypeChar (_stream);
			SAssert.ResponseType (_stream, expectedType, type, _reader);
		}

		private ArraySegment<byte> readBulk (Stream _stream)
		{
			readType (_stream, ResponseType.Bulk);
			var length = _reader.ReadNumberLine (_stream);
			byte[] _buffer = new byte[BufferConfig.EncodingBufferSize];
			// if fit in encoder buffer, use that
			if (length <= _buffer.Length) {
				_reader.ReadBulk (_stream, _buffer, 0, length);
				return new ArraySegment<byte> (_buffer, 0, length);
			}

			return new ArraySegment<byte> (_reader.ReadBulk (_stream, length));
		}

		private ArraySegment<byte>? readBulkOrNull (Stream _stream)
		{
			readType (_stream, ResponseType.Bulk);
			var length = _reader.ReadNumberLine (_stream);

			if (length == -1)
				return null;
			byte[] _buffer = new byte[BufferConfig.EncodingBufferSize];
			if (length <= _buffer.Length) {
				_reader.ReadBulk (_stream, _buffer, 0, length);
				return new ArraySegment<byte> (_buffer, 0, length);
			}

			return new ArraySegment<byte> (_reader.ReadBulk (_stream, length));
		}
	}
}
